package com.zendesk.maxwell.support;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.lmax.disruptor.BatchEventProcessor;
import com.lmax.disruptor.BlockingWaitStrategy;
import com.lmax.disruptor.EventHandler;
import com.lmax.disruptor.FatalExceptionHandler;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.Sequence;
import com.lmax.disruptor.SequenceBarrier;
import com.lmax.disruptor.WorkHandler;
import com.lmax.disruptor.WorkerPool;
import com.lmax.disruptor.dsl.Disruptor;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.LockSupport;

/**
 * @Auther: miaoguoxin
 * @Date: 2021/12/14 17:25
 * @Description:
 */
public class Test {
	private static final AtomicInteger a = new AtomicInteger(0);
	private static final int maxFullTimes = 10;
	private RingBuffer<DisEvent> msgBuffer = RingBuffer.createSingleProducer(
			new MessageEventFactory(), 1024 * 4, new BlockingWaitStrategy());
	private ExecutorService parserExecutor;
	private ExecutorService stageExecutor;
	private BatchEventProcessor<DisEvent> sinkStoreStage;
	private WorkerPool<DisEvent> workerPool;
	private DisruptorEventBus eventBus;

	//ringbuffer大小，必须是2的N次方
	private final static int RING_BUFFER_SIZE = 1024 * 4;

	public static void main(String[] args) throws InterruptedException {
		Test test = new Test();
		test.start();

		for (int i = 0; i < 1; i++) {
			for (int j = 0; j <= 1000000; j++) {
				DisEvent disEvent = new DisEvent("{\"test\":123,\"a\":\"ffff\"}", System.currentTimeMillis());
				disEvent.setI(j);
				test.publish(disEvent);
			}
		}
		Thread.sleep(3000L);
	}

	public void publish(DisEvent disEvent) {
		boolean interupted = false;
		long blockingStart = 0L;
		int fullTimes = 0;
		long next = msgBuffer.next();
		DisEvent ringBuffer = msgBuffer.get(next);
		ringBuffer.setJson(disEvent.getJson());
		ringBuffer.setI(disEvent.getI());
		ringBuffer.setTime(disEvent.getTime());
		msgBuffer.publish(next);
	}

	private void applyWait(int fullTimes) {
		int newFullTimes = Math.min(fullTimes, maxFullTimes);
		if (fullTimes <= 3) { // 3次以内
			Thread.yield();
		} else { // 超过3次，最多只sleep 1ms
			LockSupport.parkNanos(100 * 1000L * newFullTimes);
		}

	}

	public void start() throws InterruptedException {
		int tc = 2;
		this.parserExecutor = Executors.newFixedThreadPool(tc, r -> {
			Thread t = new Thread(Thread.currentThread().getThreadGroup(), r, "t", 0);
			t.setDaemon(true);
			if (t.getPriority() != Thread.NORM_PRIORITY) {
				t.setPriority(Thread.NORM_PRIORITY);
			}
			return t;
		});
		this.stageExecutor = Executors.newFixedThreadPool(2, r -> {
			Thread t = new Thread(Thread.currentThread().getThreadGroup(), r, "t1", 0);
			t.setDaemon(true);
			if (t.getPriority() != Thread.NORM_PRIORITY) {
				t.setPriority(Thread.NORM_PRIORITY);
			}
			return t;
		});

		FatalExceptionHandler handler = new FatalExceptionHandler();
//		SequenceBarrier sequenceBarrier = msgBuffer.newBarrier();
//		simpleParserStage = new BatchEventProcessor<>(msgBuffer,
//				sequenceBarrier,
//				new GateStage());
//		simpleParserStage.setExceptionHandler(handler);
//		msgBuffer.addGatingSequences(simpleParserStage.getSequence());

		SequenceBarrier dmlParserSequenceBarrier = msgBuffer.newBarrier();
		WorkHandler<DisEvent>[] workHandlers = new ParserStage[tc];
		for (int i = 0; i < tc; i++) {
			workHandlers[i] = new ParserStage();
		}
		workerPool = new WorkerPool<>(msgBuffer,
				dmlParserSequenceBarrier,
				handler,
				workHandlers);
		Sequence[] sequence = workerPool.getWorkerSequences();
		msgBuffer.addGatingSequences(sequence);

		SequenceBarrier sinkSequenceBarrier = msgBuffer.newBarrier(sequence);
		sinkStoreStage = new BatchEventProcessor<>(msgBuffer,
				sinkSequenceBarrier,
				new LastStage(System.currentTimeMillis()));
		sinkStoreStage.setExceptionHandler(handler);
		msgBuffer.addGatingSequences(sinkStoreStage.getSequence());

		//stageExecutor.submit(simpleParserStage);
		stageExecutor.submit(sinkStoreStage);
		workerPool.start(parserExecutor);
	}

	private static void pub(Disruptor<DisEvent> disruptor, int i) {
		RingBuffer<DisEvent> ringBuffer = disruptor.getRingBuffer();
		long sequence = ringBuffer.next();
		try {
			DisEvent disEvent = ringBuffer.get(sequence);
			disEvent.setI(i);
			disEvent.setJson(disEvent.getJson());
		} finally {
			ringBuffer.publish(sequence);
		}
	}

	private class GateStage implements EventHandler<DisEvent> {

		@Override
		public void onEvent(DisEvent event, long sequence, boolean endOfBatch) throws Exception {
			if (event.getI() % 10 == 0) {
				//System.out.println(event.getI());
			}
		}
	}

	private class ParserStage implements WorkHandler<DisEvent> {

		@Override
		public void onEvent(DisEvent event) throws Exception {
			JSONObject jsonObject = JSON.parseObject(event.getJson());
			event.setJsonObj(jsonObject);
			//System.out.println(event.getI());
		}
	}

	private class LastStage implements EventHandler<DisEvent> {

		private long time;

		public LastStage(long time) {
			this.time = time;
		}

		@Override
		public void onEvent(DisEvent event, long sequence, boolean endOfBatch) throws Exception {
			//System.out.println("last:" + event.getI());
			int andIncrement = a.getAndIncrement();
			if (andIncrement !=event.getI()) {
				System.out.println("顺序不对");
			}
			if (event.getI() == 1000000) {
				System.out.println("线程:" + Thread.currentThread());
				System.out.println(System.currentTimeMillis() - time);
			}
		}
	}
}
